package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradePaymentBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 从Kafka读取交易域支付成功主题数据，统计支付成功独立用户数和首次支付成功用户数
 */
public class DwsTradePaymentSucWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradePaymentSucWindow().start(10027,
                4,
                "dws_trade_payment_suc_window",
                Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 从 Kafka支付成功明细主题读取数据
        // 转换数据结构 String 转换为 JSONObject。
        KeyedStream<JSONObject, String> mapDs = stream.map(JSON::parseObject)
                // 按照 user_id 分组
                .keyBy(jsonObject -> jsonObject.getString("user_id"));

        SingleOutputStreamOperator<TradePaymentBean> processDs = mapDs.process(
                new KeyedProcessFunction<String, JSONObject, TradePaymentBean>() {
                    private ValueState<String> lastPayDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        RuntimeContext runtimeContext = getRuntimeContext();
                        lastPayDateState = runtimeContext.getState(
                                new ValueStateDescriptor<String>("lastPayDateStateDesc", Types.STRING)
                        );
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TradePaymentBean>.Context context, Collector<TradePaymentBean> collector) throws Exception {
                        String lastPayDate = lastPayDateState.value();
                        Long ts = jsonObject.getLong("ts") * 1000;
                        String today = DateFormatUtil.tsToDate(ts);
                        Long paymentScuUniqueUserCount = 0L;
                        Long paymentSucNewUserCount = 0L;
                        if (!today.equals(lastPayDate)) {
                            lastPayDateState.update(today);
                            paymentScuUniqueUserCount = 1L;
                            if (lastPayDate == null) {
                                paymentSucNewUserCount = 1L;
                            }
                        }
                        if (paymentScuUniqueUserCount == 1) {
                            collector.collect(
                                    TradePaymentBean.builder()
                                            .stt("")
                                            .edt("")
                                            .curDate("")
                                            .paymentSucUniqueUserCount(paymentScuUniqueUserCount)
                                            .paymentSucNewUserCount(paymentSucNewUserCount)
                                            .build()
                            );
                        }
                    }
                }
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradePaymentBean>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(120L))
        );

        SingleOutputStreamOperator<TradePaymentBean> reduceDs = processDs.windowAll(TumblingEventTimeWindows.of(Time.seconds(5L)))
                .reduce(
                        new ReduceFunction<TradePaymentBean>() {
                            @Override
                            public TradePaymentBean reduce(TradePaymentBean t1, TradePaymentBean t2) throws Exception {
                                t1.setPaymentSucNewUserCount(t1.getPaymentSucNewUserCount() + t2.getPaymentSucNewUserCount());
                                t1.setPaymentSucUniqueUserCount(t1.getPaymentSucUniqueUserCount() + t2.getPaymentSucUniqueUserCount());
                                return t1;
                            }
                        },
                        new ProcessAllWindowFunction<TradePaymentBean, TradePaymentBean, TimeWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<TradePaymentBean, TradePaymentBean, TimeWindow>.Context context, Iterable<TradePaymentBean> iterable, Collector<TradePaymentBean> collector) throws Exception {
                                TradePaymentBean bean = iterable.iterator().next();
                                bean.setStt(DateFormatUtil.tsToDateTime(context.window().getStart()));
                                bean.setEdt(DateFormatUtil.tsToDateTime(context.window().getEnd()));
                                bean.setCurDate(DateFormatUtil.tsToDate(context.window().getStart()));
                                collector.collect(bean);
                            }
                        }
                );


        // 统计每日独立支付人数和新增支付人数

        // 运用 Flink 状态编程，在状态中维护用户末次支付日期。
        // 若末次支付日期为 null，则将首次支付用户数和支付独立用户数均置为 1；否则首次支付用户数置为 0，判断末次支付日期是否为当日，如果不是当日则支付独立用户数置为 1，否则置为 0。最后将状态中的支付日期更新为当日。



        // 开窗、聚合
        // 度量字段求和，补充窗口起始时间和结束时间字段，ts 字段置为当前系统时间戳。


        reduceDs.map(
                new BeanToJsonMapFunction<>()
        ).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE, Constant.DWS_TRADE_PAYMENT_SUC_WINDOW));

        // 写出到Doris


    }
}
